Hook up CHASM Scheduler to Frontend handler #8694
Conversation
| delete(attributes, sadefs.TemporalNamespaceDivision) | ||
| delete(attributes, sadefs.TemporalSchedulePaused) |
There was a problem hiding this comment.
These shouldn't come back from visibility... they are system fields, not user fields.
Are you seeing those come back?
| delete(attributes, sadefs.TemporalNamespaceDivision) | ||
| delete(attributes, sadefs.TemporalSchedulePaused) | ||
|
|
||
| if s.Schedule.Policies == nil { |
There was a problem hiding this comment.
Why not call setNullableFields again here?
| s.Schedule.Policies.OverlapPolicy = s.overlapPolicy() | ||
| } | ||
| if !s.Schedule.GetPolicies().GetCatchupWindow().IsValid() { | ||
| // TODO - this should be set from Tweakables.DefaultCatchupWindow. |
There was a problem hiding this comment.
Why not pass this into the function?
| visibility := chasm.NewVisibility(ctx) | ||
| s.Visibility = chasm.NewComponentField(ctx, visibility) | ||
| visibility.SetSearchAttributes(ctx, req.FrontendRequest.GetSearchAttributes().GetIndexedFields()) | ||
| visibility.SetMemo(ctx, oldMemo) |
There was a problem hiding this comment.
nit:
| visibility := chasm.NewVisibility(ctx) | |
| s.Visibility = chasm.NewComponentField(ctx, visibility) | |
| visibility.SetSearchAttributes(ctx, req.FrontendRequest.GetSearchAttributes().GetIndexedFields()) | |
| visibility.SetMemo(ctx, oldMemo) | |
| visibility := chasm.NewVisibilityWithData(ctx, req.FrontendRequest.GetSearchAttributes().GetIndexedFields(), oldMemo) | |
| s.Visibility = chasm.NewComponentField(ctx, visibility) |
| @@ -0,0 +1,15 @@ | |||
| package chasm | |||
There was a problem hiding this comment.
Kinda hate that we have to put this here but this in not blocking the PR.
| return metadata.NewOutgoingContext(baseCtx, metadata.Pairs( | ||
| headers.ExperimentHeaderName, "chasm-scheduler", | ||
| )) |
There was a problem hiding this comment.
There should be a way to run the scheduler just with overriding the dynamic config without these specific headers, no?
| // as part of StartWorkflowExecution, but here it must be done separately (to | ||
| // maintain equal payload size limits between versions). | ||
| switch action := request.GetSchedule().GetAction().GetAction().(type) { | ||
| case *schedulepb.ScheduleAction_StartWorkflow: |
There was a problem hiding this comment.
I think we should also verify the memo and header size. Those are also payloads.
| func (wh *WorkflowHandler) DescribeSchedule(ctx context.Context, request *workflowservice.DescribeScheduleRequest) (_ *workflowservice.DescribeScheduleResponse, retError error) { | ||
| defer log.CapturePanic(wh.logger, &retError) | ||
|
|
||
| // Prefer CHASM scheduler if enabled. |
There was a problem hiding this comment.
Seems like we should first check if CHASM scheduler is enabled to avoid every call to the schedule going through this new code path, especially before this code has been thoroughly tested.
There was a problem hiding this comment.
We should also move the check if schedules are enabled to the top of this function (or any other function that handles schedules). Users may have disabled the feature completely for some arbitrary reason.
|
|
||
| if !wh.config.EnableSchedules(request.Namespace) { | ||
| return nil, errSchedulesNotAllowed | ||
| res, err := wh.updateScheduleCHASM(ctx, request) |
There was a problem hiding this comment.
What about handling not found errors here?
There was a problem hiding this comment.
What about request validation? (and make sure that you are not unaliasing search attributes on the CHASM path)
| trigger.ScheduledTime = timestamppb.Now() | ||
| } | ||
|
|
||
| res, err := wh.patchScheduleCHASM(ctx, request) |
There was a problem hiding this comment.
Missing validation and not found error checks.
Also don't issue these RPCs unless CHASM scheduler is enabled.
What changed?
I plan to address the TODO around LastCompletionResult's assertion in a separate follow-up PR.
How did you test it?
Potential risks